package com.zjl.SpringBoot.第21章_流式_响应式_编程.C_Reactive响应式核心;

import com.zjl.util.OutColour;
import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;

import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.LongConsumer;
import java.util.stream.Stream;

/**
 * 信号:正常/异常(取消)
 * SignalType：信号状态
 *      {@link SignalType#SUBSCRIBE}//被订阅
 *      {@link SignalType#REQUEST}//请求了n个元素
 *      {@link SignalType#CANCEL}//流被取消
 *      {@link SignalType#ON_SUBSCRIBE}//在订约时
 *      {@link SignalType#ON_NEXT}//在元素到达时
 *      {@link SignalType#ON_ERROR}//在流错误时
 *      {@link SignalType#ON_COMPLETE}//在流正常完成时
 *      {@link SignalType#AFTER_TERMINATE}//在流中断后
 *      {@link SignalType#CURRENT_CONTEXT}//当前上下文
 *      {@link SignalType#ON_CONTEXT}//感知上下文
 * doOnXxx API触发时机
 * 1、{@link Flux#doOnNext(Consumer)}每个数据(流的数据)到达的时候触发
 * 2、{@link Flux#doOnEach(Consumer)}每个元素(流的数据和信号)到达的时
 * 3、{@link Flux#doOnRequest(LongConsumer)}消费者请求流元素的时候
 * 4、{@link Flux#doOnError(Consumer)}流发生错误
 * 5、{@link Flux#doOnSubscribe(Consumer)}流被订阅的时候
 * 6、{@link Flux#doOnTerminate(Runnable)}发送取消/异常信号中断 了流
 * 7、{@link Flux#doOnCancel(Runnable)}流被取消
 * 8、{@link Flux#doOnDiscard(Class, Consumer)}流中元素被忽略的时候
 *
 */
public class D1_Flux高级常用 {

    @Test
    public void Flux响应式其他() throws IOException {

        Flux.range(1,7);/*  左右同理  */Flux.just(1,2,3,4,5,6,7);

        System.out.println("-----无色的是订阅者--------有色的是发布者-------------");
        //just 方法直接声明
        Flux<Integer> flux = Flux.just(1, 2, 2, 4,5)//发布者发布数据
                .delayElements(Duration.ofSeconds(1))//每 1 秒 发布一个
                .map(s -> s * s/s)//处理
                //感知类的 doOn 每个操作都是针对上1个流进行操作的
                .doOnComplete(() -> OutColour.out.printlnGreen("流正常结束了-----"))//感知 发布者
                .doOnCancel(() -> OutColour.out.printlnRed("流取消了-----"))//感知 发布者
                .doOnError(e -> OutColour.out.printlnRed("流出错了-----"))//感知 发布者
                .doOnNext(k -> OutColour.out.printlnBlue("流到达了:", k))//感知 发布者
                .doOnEach(k -> OutColour.out.printlnPurple("doOnEach-", k))//k是个 Signal 类，封装了很多
                ;
//        flux.subscribe((k) -> OutColour.out.printlnBlue("A-->>", k));//消费性接口（订阅1）
//        flux.subscribe(new MySubscriber());//使用自定义订阅者1
        flux.subscribe(new MyBaseSubscriber());//使用自定义订阅者2

        try { Thread.sleep(2000); } catch (InterruptedException e) { throw new RuntimeException(e); }


        System.in.read();//有输入就结束
    }
    @Test
    public void Flux响应式合并流() throws IOException {

        //合并流
        Flux.concat(Flux.just(1, 3, 5, 7), Flux.just(2, 4, 6))
                .subscribe(System.out::println);



        System.in.read();//有输入就结束
    }
    public static void main(String[] args) {

        System.out.println("============================");
        Mono.just(1).subscribe(System.out::print);
        //其他的方法
        Integer[] array = {1, 2, 3, 4};
        Flux.fromArray(array);

        List<Integer> list = Arrays.asList(array);
        Flux.fromIterable(list);

        Stream<Integer> stream = list.stream();
        Flux.fromStream(stream);//讲普通流转为  响应式流

    }

}
class MySubscriber implements Subscriber<Integer>{

    Subscription subscription;

    @Override
    public void onSubscribe(Subscription subscription) {
        System.out.println("订阅者和发布者绑定好了：" + subscription);
        this.subscription = subscription;
        this.subscription.request(1);
    }

    @Override
    public void onNext(Integer integer) {
        System.out.println("元素到达：" + integer);
        this.subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("异常出错：" + throwable);

    }

    @Override
    public void onComplete() {
        System.out.println("留结束了---");
    }
}
